package tech.realcpf.duckengine1.sqlparser;

import org.apache.calcite.sql.SqlIdentifier;
import org.apache.calcite.sql.SqlNode;
import org.apache.calcite.sql.SqlOrderBy;
import org.apache.calcite.sql.SqlSelect;
import tech.realcpf.duckengine1.biz.BizCatalog;
import tech.realcpf.duckengine1.biz.BizInstanceException;

import java.util.List;
import java.util.Locale;
import java.util.Map;

public class TableNameSqlReLocator implements SqlReLocator {

    private final Map<String, BizCatalog> catalogMap;
    private final String s3BizUserUrl;

    public TableNameSqlReLocator(Map<String, BizCatalog> catalogMap, String s3BizUserUrl) {
        this.catalogMap = catalogMap;
        this.s3BizUserUrl = s3BizUserUrl;
    }

    @Override
    public void relocate(SqlNode sqlNode) {
        SqlNode sqlQuery;
        if (sqlNode instanceof SqlOrderBy orderBy) {
            sqlQuery = orderBy.query;
        } else {
            sqlQuery = sqlNode;
        }
        if (sqlQuery instanceof SqlSelect select) {
            SqlNode sqlFrom = select.getFrom();
            if (sqlFrom instanceof SqlIdentifier sqlFromName) {
                String catalogName = sqlFromName.names.get(0);
                String tableName = sqlFromName.names.get(1);
                if (!catalogMap.containsKey(catalogName)) {
                    throw new BizInstanceException("catalog " + catalogName + " not exits.");
                }
                BizCatalog catalog = catalogMap.get(catalogName);
                String fileExt = catalog.getFileExt().toLowerCase(Locale.ROOT);
                String s3TablePath = s3BizUserUrl + "/" + catalogName + "/" + tableName + "/" + tableName + "." + fileExt;
                sqlFromName.setNames(List.of(s3TablePath), List.of(sqlFromName.getParserPosition()));
            }
        }
    }

    @Override
    public boolean support(SqlNode sqlNode) {
        return sqlNode instanceof SqlSelect || sqlNode instanceof SqlOrderBy;
    }
}
